package cn.bby.query.server.domain.service.update;

import cn.bby.query.base.constant.DataChangeOpType;
import cn.bby.query.server.domain.agg.metadata.entity.DsTableInfo;
import cn.bby.query.server.domain.agg.metadata.entity.InterfaceInfo;
import cn.bby.query.server.domain.agg.metadata.entity.valobj.ColumnInfo;
import cn.bby.query.server.domain.agg.metadata.entity.valobj.TableInfo;
import cn.bby.query.server.domain.agg.metadata.repository.DsSettingsRepository;
import cn.bby.query.server.domain.agg.metadata.repository.InterfaceInfoRepository;
import cn.bby.query.server.domain.service.dto.DataChangeEvent;
import cn.bby.query.server.domain.service.dto.DataUpdateDetail;
import cn.bby.query.server.domain.service.dto.DataUpdateParam;
import cn.bby.query.server.domain.service.fetcher.DataFetcher;
import cn.bby.query.server.domain.service.manager.DataStorageManager;
import cn.bby.query.server.domain.service.manager.DelayEventBufManager;
import cn.bby.query.server.domain.share.constant.InterfaceStatus;
import cn.bby.query.server.domain.share.util.ErrorLogUtil;
import com.alibaba.fastjson2.JSONObject;
import jakarta.annotation.Resource;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author ben
 * @since 2025/4/30 23:59
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class DataUpdateHandleServiceImpl implements DataUpdateHandleService {

    private final DsSettingsRepository dsSettingsRepository;
    private final InterfaceInfoRepository interfaceInfoRepository;
    @Resource
    private ApplicationContext applicationContext;
    @Resource
    private DataStorageManager dataStorageManager;
    @Resource
    private DelayEventBufManager delayEventBufManager;
    @Resource
    private ErrorLogUtil errorLogUtil;

    @Override
    public void handleUpdateDataEvent(List<DataChangeEvent> changeEvents) {
        Map<String, List<DataChangeEvent>> tableEventMap = changeEvents.stream()
                .collect(Collectors.groupingBy(DataChangeEvent::getTableFullName));

        Collection<DsTableInfo> tableInfos = dsSettingsRepository.getTableInfoByNames(tableEventMap.keySet());
        if (tableInfos.isEmpty()) {
            log.warn("未找到表信息，忽略更新:{}", tableEventMap.keySet());
            return;
        }
        Map<Long, String> tableIdNameMap = tableInfos.stream()
                .collect(Collectors.toMap(DsTableInfo::getTableId, DsTableInfo::getTableName));
        Map<String, Long> tableNameIdMap = tableIdNameMap.entrySet().stream()
                .collect(Collectors.toMap(Map.Entry::getValue, Map.Entry::getKey));
        Collection<InterfaceInfo> interfaceInfos = interfaceInfoRepository.getByTableIds(tableIdNameMap.keySet());

        if (interfaceInfos.isEmpty()) {
            log.warn("未找到接口信息，忽略更新:{}", tableEventMap.keySet());
            return;
        }
        // map：k-interfaceId,v-tableNames
        Map<InterfaceInfo, List<DataChangeEvent>> interfaceIdEventMap = interfaceInfos.stream()
                .collect(Collectors.toMap(Function.identity(), interfaceInfo -> interfaceInfo.getColumns()
                        .stream()
                        .flatMap(columnInfo -> {
                            String tabName = tableIdNameMap.get(columnInfo.getTableId());
                            if (tabName == null) {
                                log.warn("未找到表信息，忽略更新:{}", tabName);
                                return null;
                            }

                            return tableEventMap.get(tabName).stream();
                        })
                        .filter(Objects::nonNull)
                        .distinct()
                        .collect(Collectors.toList())
                ));
        if (interfaceIdEventMap.isEmpty()) {
            log.warn("未找到接口信息，忽略更新:{}", tableEventMap.keySet());
            return;
        }
        interfaceIdEventMap.forEach((interfaceInfo, events) -> {
            List<DataUpdateDetail> updateDetails = events.stream()
                    .map(e -> buildDetail(interfaceInfo, tableNameIdMap, e))
                    .filter(e -> needHandle(interfaceInfo, e))
                    .collect(Collectors.toList());
            if (updateDetails.isEmpty()) {
                log.debug("未找到数据，忽略更新:{}", events);
                return;
            }
            if (!InterfaceStatus.READY.equals(interfaceInfo.getStatus())) {
                log.info("接口未就绪，将数据保存至延迟事件缓冲区:{}", interfaceInfo.getInterfaceName());
                delayEventBufManager.createEvents(interfaceInfo.getInterfaceId(), updateDetails);
                return;
            }
            try {
                Map<Boolean, List<DataUpdateDetail>> canDirectUpdateMap = updateDetails.stream()
                        .collect(Collectors.partitioningBy(event -> canDirectUpdate(interfaceInfo, event)));

                if (!canDirectUpdateMap.get(true).isEmpty()) {
                    doDirectUpdateBatch(interfaceInfo, canDirectUpdateMap.get(true));
                }
                if (!canDirectUpdateMap.get(false).isEmpty()) {
                    doReLoadBatch(interfaceInfo, canDirectUpdateMap.get(false));
                }
            } catch (Exception e) {
                errorLogUtil.write(interfaceInfo.getInterfaceId(), "数据更新失败,存储导缓冲区", e);
                delayEventBufManager.createEvents(interfaceInfo.getInterfaceId(), updateDetails);
            }
        });
    }

    private DataUpdateDetail buildDetail(InterfaceInfo interfaceInfo, Map<String, Long> tableNameIdMap, DataChangeEvent e) {
        DataUpdateDetail detail = new DataUpdateDetail()
                .setAfter(e.getAfter())
                .setBefore(e.getBefore())
                .setOpType(e.getOpType())
                .setData(e.getAfter() != null ? e.getAfter() : e.getBefore())
                .setTableId(tableNameIdMap.get(e.getTableFullName()));
        if (detail.getData() != null) {
            ColumnInfo column = interfaceInfo.getIdentityColumnByTableId(detail.getTableId());
            detail.setIdentityVal(detail.getData().getString(column.getFieldName()));
            detail.setIdentityCol(column);
        }
        return detail;
    }

    private boolean needHandle(InterfaceInfo interfaceInfo, DataUpdateDetail detail) {
        TableInfo table = interfaceInfo.getSubTableById(detail.getTableId());
        if (table == null || detail.getIdentityVal() == null) {
            return false; // 不依赖这表/沒有定位值
        }
        JSONObject after = detail.getAfter();
        JSONObject before = detail.getBefore();
        if ((after != null && !after.isEmpty()) || (before != null && !before.isEmpty())) {
            return false;
        }

        return table.getIsComplex(); // 复杂依赖都要处理
    }

    // 是否可以直接更新
    private boolean canDirectUpdate(InterfaceInfo interfaceInfo, DataUpdateDetail detail) {
        TableInfo table = interfaceInfo.getSubTableById(detail.getTableId());
        if (table.getIsComplex()) {
            return false; // 复杂依赖重新加载
        }
        if (DataChangeOpType.DELETE.equals(detail.getOpType())) {
            // 非复杂的可直接删除
            return true;
        }
        if (DataChangeOpType.CREATE.equals(detail.getOpType())) {
            // 非主表创建可忽略
            return !Objects.equals(detail.getTableId(), interfaceInfo.getMainTableId());
        }
        // 比较变化的数据
        JSONObject after = detail.getAfter();
        JSONObject before = detail.getBefore();
        if (after == null || before == null) {
            // 没有给变化详情
            return false;
        }
        List<String> columns = interfaceInfo.getCompColsByTableId(detail.getTableId());
        for (String column : columns) {
            if (!Objects.equals(after.get(column), before.get(column))) {
                // 复杂字段有变化
                return false;
            }
        }
        return true;
    }

    private void doDirectUpdateBatch(InterfaceInfo interfaceInfo, List<DataUpdateDetail> details) {
        Map<String, List<DataUpdateDetail>> opTypeMap = details.stream()
                .collect(Collectors.groupingBy(DataUpdateDetail::getOpType));
        // 删除的
        Map<ColumnInfo, Set<String>> deletedFieldValMaps = opTypeMap.getOrDefault(DataChangeOpType.DELETE, Collections.emptyList())
                .stream()
                .collect(Collectors.groupingBy(
                        DataUpdateDetail::getIdentityCol,
                        Collectors.mapping(DataUpdateDetail::getIdentityVal, Collectors.toSet())
                ));

        // 创建的不管
        // 更新的
        List<DataUpdateParam> updateParams = opTypeMap.getOrDefault(DataChangeOpType.UPDATE, Collections.emptyList())
                .stream()
                .map(detail -> {
                    ColumnInfo identityCol = detail.getIdentityCol();
                    String identityVal = detail.getIdentityVal();

                    if (identityCol == null || identityVal == null) {
                        return null; // 没有定位信息，无法更新
                    }
                    JSONObject after = detail.getAfter();
                    JSONObject before = detail.getBefore();

                    if (after == null || before == null) {
                        return null; // 如果没有提供完整的变化数据，跳过
                    }

                    // 获取该表的所有列名
                    List<ColumnInfo> columns = interfaceInfo.getSimpColsByTableId(detail.getTableId());
                    Map<ColumnInfo, String> updateCols = new HashMap<>();
                    // 检查每个字段是否发生变化，并构建变化字段的映射
                    boolean hasChanges = false;
                    for (ColumnInfo column : columns) {
                        String fieldName = column.getFieldName();
                        String afterValue = after.getString(fieldName);
                        String beforeValue = before.getString(fieldName);

                        if (!Objects.equals(afterValue, beforeValue)) {
                            updateCols.put(column, afterValue);
                            hasChanges = true;
                        }
                    }
                    DataUpdateParam param = new DataUpdateParam()
                            .setUpdateCols(updateCols)
                            .setIdentityCol(detail.getIdentityCol())
                            .setIdentityVal(detail.getIdentityVal());
                    return hasChanges ? param : null;
                }).filter(Objects::nonNull)
                .toList();

        // 执行删除操作
        if (!deletedFieldValMaps.isEmpty()) {
            dataStorageManager.deleteData(interfaceInfo, deletedFieldValMaps);
        }

        // 执行更新操作（如果有）
        if (!updateParams.isEmpty()) {
            // 这里可能需要根据具体情况决定如何执行更新
            dataStorageManager.updateDataByParam(interfaceInfo, updateParams);
        }
    }

    private void doReLoadBatch(InterfaceInfo interfaceInfo, List<DataUpdateDetail> dataUpdateDetails) {
        if (dataUpdateDetails.isEmpty()) {
            return;
        }
        Map<ColumnInfo, Set<String>> fieldValMaps = new HashMap<>();
        Set<String> ids = new HashSet<>();
        Long mainTableId = interfaceInfo.getMainTableId();
        for (DataUpdateDetail detail : dataUpdateDetails) {
            Long tableId = detail.getTableId();
            JSONObject data = detail.getData();
            ColumnInfo priCol = interfaceInfo.getIdentityColumnByTableId(tableId);
            if (priCol == null) {
                log.info("未找到主键列信息，忽略更新:{}", tableId);
                continue;
            }
            String priVal = data.getString(priCol.getLinkCol());
            if (priVal == null) {
                log.info("未找到主键值，忽略更新:{}", tableId);
                continue;
            }
            if (mainTableId != null && mainTableId.equals(tableId)) {
                ids.add(priVal);
                continue;
            }
            fieldValMaps.computeIfAbsent(priCol, k -> new HashSet<>()).add(priVal);
        }
        if (!fieldValMaps.isEmpty()) {
            ids.addAll(dataStorageManager.queryIdByPrimaryValList(interfaceInfo, fieldValMaps));
        }
        if (ids.isEmpty()) {
            log.debug("全部未找到主键：{}", dataUpdateDetails);
            return;
        }
        DataFetcher fetcher = applicationContext.getBean("KeyFullDataFetcher." + interfaceInfo.getPrimaryColumnType(), DataFetcher.class);
        fetcher.reFetchData(interfaceInfo, ids);
    }

    @Scheduled(fixedDelay = 30, timeUnit = TimeUnit.SECONDS)
    @Override
    public void handleDelayEvents() {
        Map<Long, List<DataUpdateDetail>> interfaceMap = delayEventBufManager.popReadyEvents(2000);
        if (interfaceMap.isEmpty()) {
            return;
        }
        log.info("处理延迟事件数量:{}", interfaceMap.size());
        interfaceMap.forEach((interfaceId, events) -> {
            InterfaceInfo interfaceInfo = interfaceInfoRepository.getById(interfaceId);
            doDirectUpdateBatch(interfaceInfo, events);
        });
    }
}
